package com.platform.study.thread.executor;

import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * 并行计算数组的和
 * ExecutoreService提供了submit()方法，传递一个Callable，或Runnable，返回Future。
 * 如果Executor后台线程池还没有完成Callable的计算，这调用返回Future对象的get()方法，会阻塞直到计算完成。
 * <p/>
 * User: wangyanginfo
 * Date: 13-2-1
 * Time: 下午3:20
 */
public class ConcurrentCalculator {

    private static Logger logger = Logger.getLogger(ConcurrentCalculator.class);

    private ExecutorService exec;
    private int cpuCoreNumber;
    private List<Future<Long>> tasks = new ArrayList<Future<Long>>();

    // 内部类
    class SumCalculator implements Callable<Long> {
        private int[] numbers;
        private int start;
        private int end;

        public SumCalculator(final int[] numbers, int start, int end) {
            this.numbers = numbers;
            this.start = start;
            this.end = end;
        }

        public Long call() throws Exception {
            Long sum = 0l;
            for (int i = start; i < end; i++) {
                sum += numbers[i];
            }
            return sum;
        }
    }

    public ConcurrentCalculator() {
        cpuCoreNumber = Runtime.getRuntime().availableProcessors();
        logger.debug("cpuCoreNumber：" + cpuCoreNumber);
        exec = Executors.newFixedThreadPool(cpuCoreNumber);
    }

    public Long sum(final int[] numbers) {
        // 根据CPU核心个数拆分任务，创建FutureTask并提交到Executor
        for (int i = 0; i < cpuCoreNumber; i++) {
            int increment = numbers.length / cpuCoreNumber + 1;
            int start = increment * i;
            int end = increment * i + increment;
            if (end > numbers.length)
                end = numbers.length;
            SumCalculator subCalc = new SumCalculator(numbers, start, end);
            FutureTask<Long> task = new FutureTask<Long>(subCalc);
            tasks.add(task);
            if (!exec.isShutdown()) {
                exec.submit(task);
            }
        }
        return getResult();
    }

    /**
     * 迭代每个只任务，获得部分和，相加返回
     *
     * @return
     */
    public Long getResult() {
        Long result = 0l;
        for (Future<Long> task : tasks) {
            try {
                // 如果计算未完成则阻塞
                Long subSum = task.get();
                result += subSum;
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
        return result;
    }

    public void close() {
        exec.shutdown();
    }

    public static void main(String[] args) {
        int len = 100000000;
        int[] numbers = new int[len];
        for (int i = 0; i < len; i++) {
            numbers[i] = i + 1;
        }
        long startTime = System.currentTimeMillis();
        ConcurrentCalculator calc = new ConcurrentCalculator();
        Long sum = calc.sum(numbers);
        logger.debug(sum);
        calc.close();
        logger.debug(System.currentTimeMillis() - startTime);

        startTime = System.currentTimeMillis();
        Long sum1 = 0L;
        for (int i = 0; i < len; i++) {
            sum1 += numbers[i];
        }
        logger.debug(sum);
        logger.debug(System.currentTimeMillis() - startTime);
    }

}
